package com.atguigu.spx;

import com.atguigu.spx.bean.Job;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.print.attribute.standard.JobSheets;
import java.sql.*;
import java.util.*;

/**
 * create by Shipeixin on  2021-05-26  16:04
 */
public class CustomProducer {

    private static Connection connection;

    private static PreparedStatement prepareStatement;



    public static void main(String[] args) throws SQLException, InterruptedException {


        Properties properties = new Properties();

        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.10.103:9092");

        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> preoducer = new KafkaProducer<>(properties);

        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/kafka_test?characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2B8","root","root");

        prepareStatement = connection.prepareStatement("select * from `kafka_source`");

        boolean hasRes = prepareStatement.execute();
        List<LinkedHashMap<String, String>> result = null;
        if (hasRes){
            ResultSet resultSet = prepareStatement.getResultSet();
            ResultSetMetaData metaData = resultSet.getMetaData();

            int columnCount = metaData.getColumnCount();
            List<String> columnNames = new ArrayList<>();
            for (int i = 1; i <= columnCount; i++){
                columnNames.add(metaData.getColumnName(i));
            }
            result = getResult(resultSet, columnNames);
        }

        if (result != null){
            for (LinkedHashMap<String, String> res : result) {
                preoducer.send(new ProducerRecord<>("forth", res.toString()));
            }
        }

        preoducer.close();
    }




    public static List<LinkedHashMap<String,String>> getResult(ResultSet resultSet, List<String> columnNames) throws SQLException {
        List<LinkedHashMap<String,String>> res = new ArrayList<>();
        while (resultSet.next()){
            LinkedHashMap<String, String> row = new LinkedHashMap<>();
            for (String columnName : columnNames) {
                String value = resultSet.getString(columnName);
                row.put(columnName, value);
            }
            res.add(row);
        }
        return res;
    }

}
